package com.bw.gmall.realtime.utils;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

//jdbc    hbase    Phoenix 工具类类似于mysql 操作 hbase
//自定义sink  还记得吗？？？？？
public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
    //线程池  数据源元
    private DruidDataSource druidDataSource;


    //初始化方法只加载一次
    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource=DruidDSUtil.createDataSource();

    }

    //invoke  来一条数据就执行一次
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        /*
         * {"sinkTable":"dim_sku_info",
         * "database":"gmall",
         * "xid":165935,
         * "data":{"id":36},
         * "commit":true,"type":"insert","table":"sku_info","ts":1592808490}
         * */


        //获取连接   工具类  连接池
        DruidPooledConnection connection = druidDataSource.getConnection();
        //获取hbase表名  upsert    表名  values
        String sinkTable = value.getString("sinkTable");

        //获取字段
        JSONObject jsonObject = value.getJSONObject("data");

        //data:{"id":1,"name":"zs"}
        // upsert   into  表名（字段名称）  values(值)

        //获取数据类型
//        String type = value.getString("type");
        //如果为更新数据,则需要删除Redis中的数据
//        if ("update".equals(type)) {
//            DimUtil.delDimInfo(sinkTable.toUpperCase(), jsonObject.getString("id"));
//        }
        //["id","name"] --->  用,隔开   id,name
        // insert  user_info(id,name)
        //['1','zs']    用','隔开
        //insert  user_info(id,name)  values('1','zs')
        //{"name":'zs',"age":12}
        Set<String> keySet = jsonObject.keySet();  //id,name



        Collection<Object> values1 = jsonObject.values(); // 1,zs
        //[1,2,3,4,5,6]  scala    arr.mkstring    '1','2','3','4','5,6','78','9'
        //StringUtils.join(values1,",")  ===>     values1.mkstring(",")
        String  sql="upsert  into "+ GmallConfig.HBASE_SCHEMA+"."+sinkTable+
                "("+ StringUtils.join(keySet,",")+") values('"
                //(10001,zs,12)  明显错误  （'1001','zs','12'）
                +StringUtils.join(values1,"','")+"')";
        System.out.println(sql);
        //编译sql
        PreparedStatement ps = connection.prepareStatement(sql);


        //执行
        ps.execute();
        connection.commit();


        //释放连接
        connection.close();
    }
}
